#include "ChannelBuffer.h"
#include <assert.h>
#ifndef IS_WINDOWS
  #include <errno.h>
  #include <memory.h>
  #include <sys/uio.h> // for iovec
#endif
#include "TcpSocketOps.h"
#include "Options.h"
#include "Log.h"
using std::string;
namespace dsth {

const int ChannelBuffer::kPrependSize = 8;
const int ChannelBuffer::kInitSize = 1024;

ChannelBuffer::ChannelBuffer()
    : begin_(0), end_(0), writer_(0), reader_(0) {
    int buf_len = GetOption("ChannelBufferLength").intVal;
    int pre_len = kPrependSize;
    begin_ = static_cast<char*>(::malloc(buf_len + pre_len));
    writer_ = reader_ = begin_ + pre_len;
    end_ = writer_ + buf_len;
}

ChannelBuffer::ChannelBuffer(size_t initSize) 
    : begin_(0), end_(0), writer_(0), reader_(0) {
    size_t buf_len = initSize;
    size_t pre_len = kPrependSize;
    begin_ = static_cast<char*>(::malloc(buf_len + pre_len));
    writer_ = reader_ = begin_ + pre_len;
    end_ = writer_ + buf_len;
}

ChannelBuffer::~ChannelBuffer() {
    ::free(begin_);
}

string ChannelBuffer::FetchAsString() {
    string str(reader_, writer_ - reader_);
    RetrieveAll();
    return str;
}

int32_t ChannelBuffer::FetchAsInt32() {
    assert(ReadableBytes() >= sizeof(int32_t));
    int32_t res = PeekInt32();
    Retrieve(sizeof res);
    return net::N2H_32(res);
}

int32_t ChannelBuffer::PeekInt32() {
    assert(ReadableBytes() >= sizeof(int32_t));
    int32_t res = 0;
    memcpy(&res, reader_, sizeof res);
    return net::N2H_32(res);
}

void ChannelBuffer::EnsureCapacity(size_t len) {
    if (WritableBytes() < len) {
        IncreaseCapacity(len);
    }
}

void ChannelBuffer::HasWritten(size_t len) {
    writer_ += len;
    assert(writer_ <= end_);
}

void ChannelBuffer::IncreaseCapacity(size_t more) {
    size_t new_size = end_ - begin_ + more;
    size_t perwrited = writer_ - begin_;
    size_t perread = reader_ - begin_;
    char* new_buf = static_cast<char*>(::realloc(begin_, new_size));
    if (NULL == new_buf) {
        ::free(begin_);
        LOG_FATAL("[ChannelBuffer::IncreaseCapacity] out of memory. current size:%d", end_ - begin_);
    } else if (new_buf != begin_) {
        begin_ = new_buf;
        writer_ = begin_ + perwrited;
        reader_ = begin_ + perread;
        end_ = begin_ + new_size;
        assert(end_ >= (writer_ + more));
    }
    else {
        end_ = begin_ + new_size;
    }
}

void ChannelBuffer::Perpend(const void* data, size_t len) {
    assert(len <= PrependableBytes());
    reader_ -= len;
    const char* d = static_cast<const char*>(data);
    memcpy(reader_, d, len);
}

void ChannelBuffer::AppendInt32(int32_t x) {
    int32_t be32 = net::H2N_32(x);
    Append(&be32, sizeof be32);
}

void ChannelBuffer::AppendInt16(int16_t x) {
    int16_t be16 = htons(x);
    Append(&be16, sizeof be16);
}

void ChannelBuffer::Append(const void* data, size_t len) {
    Append(static_cast<const char*>(data), len);
}

void ChannelBuffer::Append(const char* data, size_t len) {
    assert(data && len > 0);
    size_t writable = WritableBytes();
    int more = static_cast<int>(len - writable);
    if (more > 0) {
        size_t frontSize = reader_ - begin_ - kPrependSize;
        if (len < frontSize) {
            LOG_DEBUG("[ChannelBuffer::Append] Move Memory, frontSize:%d, Writable:%d", frontSize, writable);
            size_t cpySize = writer_ - reader_;
            memcpy(begin_ + kPrependSize, reader_, cpySize);
            reader_ = begin_ + kPrependSize;
            writer_ = reader_ + cpySize;
        }
        else {
            IncreaseCapacity(more);
            /*
            LOG_DEBUG("[ChannelBuffer::Append] Append Memory len:%d, more:%d", len, more);
            size_t new_size = end_ - begin_ + more;
            size_t perwrited = writer_ - begin_;
            size_t perread = reader_ - begin_;
            char* new_buf = static_cast<char*>(::realloc(begin_, new_size));
            if (NULL == new_buf) {
                ::free(begin_);
                LOG_FATAL("[ChannelBuffer::Append] out of memory. current size:%d", end_ - begin_);
            } else if (new_buf != begin_) {
                LOG_DEBUG("[ChannelBuffer::Append] new_size:%d, perwrited:%d, perread:%d\n", new_size, perwrited, perread);
                begin_ = new_buf;
                writer_ = begin_ + perwrited;
                reader_ = begin_ + perread;
                end_ = begin_ + new_size;
                assert(end_ >= (writer_ + more));
            }
            else {
                end_ = begin_ + new_size;
            }*/
        }
    }
    memcpy(writer_, data, len);
    writer_ += len;
    assert(end_ >= writer_);
}

int ChannelBuffer::ReadFd(int fd, /*out*/int* err) {
    char extra_buf[65536];
    size_t writable_bytes = WritableBytes();
#ifdef IS_WINDOWS
    WSABUF iovec[2];
    iovec[0].buf = writer_;
    iovec[0].len = static_cast<u_long>(writable_bytes);
    iovec[1].buf = extra_buf;
    iovec[1].len = sizeof extra_buf;
    int recv_bytes, flag = 0;
    int ret =  net::Readv(fd, iovec, 2, &recv_bytes, &flag);
    LOG_TRACE("[ChannelBuffer::ReadFd] recv_bytes:%d writable_byte:%d fd:%d.", recv_bytes, writable_bytes, fd);
    if (ret == SOCKET_ERROR) {
        *err = ::WSAGetLastError();
        return ret;
    }
    else if (recv_bytes <= static_cast<int>(writable_bytes)) {
#else // POSIX
    struct iovec iovec[2];
    iovec[0].iov_base = writer_;
    iovec[0].iov_len = writable_bytes;
    iovec[1].iov_base = extra_buf;
    iovec[1].iov_len = sizeof extra_buf;
    ssize_t recv_bytes = net::Readv(fd, iovec, 2);
    if (recv_bytes <= 0) {
        //TODO: fixme
        *err = errno;
        LOG_ERROR("[ChannelBuffer::ReadFd] socket readv, errorcode:%d.", errno);
        return recv_bytes;
    }
    else if (recv_bytes <= static_cast<ssize_t>(writable_bytes)) {
#endif
        writer_ += recv_bytes;
    }
    else {
        // need memory move
        writer_ = end_;
        Append(extra_buf, recv_bytes - writable_bytes);
    }
    return recv_bytes;
}

int ChannelBuffer::WriteFd(int fd) {
    return net::Write(fd, reader_, ReadableBytes());
}

} // namespace
